package com.hisense.hiioc.hismartplate.server;

import com.hisense.hiioc.hismartplate.utils.EmptyUtils;
import com.hisense.hiioc.hismartplate.utils.NettyUtils;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @Author: almo
 * @Date: 2018/12/5 下午3:39
 * @Version 1.0
 */

public class CoreManager {

    private static final Logger logger = LoggerFactory.getLogger(CoreManager.class);
    private static volatile CoreManager coreManager;

    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock(false);
    private ConcurrentMap<Channel, DevClient> devClients = new ConcurrentHashMap<>();
    private AtomicInteger clientCount = new AtomicInteger(0);

    public static CoreManager getInstance() {
        if (coreManager == null) {
            synchronized (CoreManager.class) {
                if (coreManager == null)
                    coreManager = new CoreManager();
            }
        }
        return coreManager;
    }

    private CoreManager() {

    }

    public void addChannel(Channel channel, HttpHeaders headers) {

        final String remoteAddr = NettyUtils.parseChannelRemoteAddr(channel);

        if (!channel.isActive()) {
            logger.error("the target channel is died, iPAddress:{}", remoteAddr);
        }

        final DevClient client = new DevClient();
        client.setIPAddress(remoteAddr);
        client.setChannel(channel);
        client.setTime(System.currentTimeMillis());

        client.setDeviceId(headers.get("deviceId"));
        client.setScreenId(headers.get("screenId"));

        devClients.put(channel, client);
        clientCount.incrementAndGet();
    }

    public void authorizeChannel(Channel channel) {
        //ToDo some authorized information for the channel

    }

    /**
     * remove the channel from local cache，and closeChannel
     *
     * @param channel the target channel
     */
    public void removeChannel(Channel channel) {
        try {
            logger.warn("channel will be remove, iPAddress is :{}", NettyUtils.parseChannelRemoteAddr(channel));
            rwLock.writeLock().lock();
            channel.close();
            DevClient client = devClients.get(channel);
            if (client != null) {
                DevClient tmp = devClients.remove(channel);
                if (tmp != null && tmp.isAuthorized()) {
                    // 减去一个认证用户
                    clientCount.decrementAndGet();
                }
            }
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    /**
     * broadcast ping message
     */
    void broadcastPing() {
        try {
            rwLock.readLock().lock();
            logger.info("broadCastPing user Count: {}", clientCount.intValue());
            Set<Channel> keySet = devClients.keySet();
            for (Channel ch : keySet) {
                DevClient client = devClients.get(ch);
                if (client == null || !client.isAuthorized()) continue;
                ch.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildPingProto()));
            }
        } finally {
            rwLock.readLock().unlock();
        }
    }

    /**
     * broadcast message
     *
     * @param message
     */
    public void broadcastMsg(String message, Object extension) {
        if (EmptyUtils.isNotEmpty(message)) {
            try {
                rwLock.readLock().lock();
                Set<Channel> keySet = devClients.keySet();
                for (Channel ch : keySet) {
                    DevClient clinet = devClients.get(ch);
                    if (clinet == null || !clinet.isAuthorized()) continue;
                    ch.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildMessProto(message, extension)));
                }
            } finally {
                rwLock.readLock().unlock();
            }
        }
    }

    /**
     * broadcast system message
     */
    public void broadcastSysMsg(String mess) {
        try {
            rwLock.readLock().lock();
            Set<Channel> keySet = devClients.keySet();
            for (Channel ch : keySet) {
                DevClient client = devClients.get(ch);
                if (client == null || !client.isAuthorized()) continue;
                ch.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildSystProto(mess)));
            }
        } finally {
            rwLock.readLock().unlock();
        }
    }

    /**
     * send system message to the target client according to the device Id
     *
     * @param deviceId
     * @param mess
     */
    public void sendSystemMsgByDeviceId(String deviceId, String mess) {
        devClients.forEach((channel, client) -> {
            if (client.getDeviceId().equalsIgnoreCase(deviceId)) {
                sendSystemMsg(channel, mess);
            }
        });
    }

    /**
     * send message to the specified client by screen id
     */
    public void sendMsgByScreenId(final String screenId, final String message, final Object extension) {
        devClients.forEach((channel, client) -> {
            if (client.getScreenId().equalsIgnoreCase(screenId)) {
                sendMsg(channel, message, extension);
            }
        });
    }

    /**
     * send system message by specified channel
     *
     * @param mess
     */
    private void sendSystemMsg(Channel channel, String mess) {
        channel.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildSystProto(mess)));
    }

    /**
     * send message
     *
     * @param mess
     */
    private void sendMsg(Channel channel, String mess, Object extension) {
        channel.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildMessProto(mess, extension)));
    }

    /**
     * send pong message
     *
     * @param channel the target channel
     */
    public void sendPong(Channel channel) {
        channel.writeAndFlush(new TextWebSocketFrame(ProtoBuilder.buildPongProto()));
    }

    /**
     * scan and close inactive channel
     */
    void scanNotActiveChannel() {
        Set<Channel> keySet = devClients.keySet();
        for (Channel ch : keySet) {
            DevClient client = devClients.get(ch);
            if (client == null) continue;
            if (!ch.isOpen() || !ch.isActive() || (!client.isAuthorized() &&
                    (System.currentTimeMillis() - client.getTime()) > 20000)) {
                removeChannel(ch);
            }
        }
    }

    public DevClient getDevClientByChannel(Channel channel) {
        return devClients.get(channel);
    }

    public DevClient getDevClientByScreenId(String screenID) {

        for(DevClient client:devClients.values()) {
            if(client.getScreenId().equalsIgnoreCase(screenID))
                return client;
        }
        return null;
    }

    public void updateDevClientTime(Channel channel) {
        DevClient userInfo = getDevClientByChannel(channel);
        if (userInfo != null) {
            userInfo.setTime(System.currentTimeMillis());
        }
    }

}
